package pfsimulator;

import java.util.*;

import log.PVLogger;

/**
 * We first init the network topology, then we start to build
 * ParaflowSimulator using all needed information
 */

public class ParaflowSimulator {
    private int numOfpNode;
    //private long[][] capacityGraph;
    //private long[][] flowGraph;
    private Long[] uplinkCapacity;
    private Long[] downlinkCapacity;
    private Long[] availableUplinkBw;
    private Long[] availableDownlinkBw;
    private PNode[] pNodes = null;

    //mapping vnode id to its hosting PNode
    private Map<String,PNode> vnodeIdToPNode = new HashMap<String, PNode>();
    //mapping vnode id to its hosting PNode's index in shuffle graph
    private Map<String,Integer> vnodeIdToIdx = new HashMap<String, Integer>();
    //mapping flow id to FlowRun object
    private Map<Long, FlowRun> flowActors = new HashMap<Long, FlowRun>();

    public static PVLogger LOG = PVLogger.getLogger(ParaflowSimulator.class);
    volatile private Queue<FlowFinishEvent> flowFinishEventQueue = new LinkedList<FlowFinishEvent>();

    public ParaflowSimulator() {
        super();
        this.numOfpNode = 0;
        this.uplinkCapacity = null;
        this.downlinkCapacity = null;
        this.availableDownlinkBw = null;
        this.availableUplinkBw = null;
        this.pNodes = null;
    }

    //
    // Internal functions
    //

    // Get running flow given flow id
    private FlowRun getRunningFlow(long flowId){
        return this.flowActors.get(flowId);
    }

    // Unlocked version
    private void allocate(int src, int dst, long delta){
        if(availableUplinkBw[src] < delta || availableDownlinkBw[dst] < delta){
            LOG.error("Not enough bandwidth available: " + delta + " required, (" + availableUplinkBw[src] + ","
                    + availableDownlinkBw[dst] + ") available");
        } else {
            availableUplinkBw[src] -= delta;
            availableDownlinkBw[dst] -= delta;
        }
    }

    private void deallocate(int src, int dst, long delta){
        availableUplinkBw[src] += delta;
        availableDownlinkBw[dst] += delta;
    }

    //
    // APIs
    //

    /**
     *  Construct using the given list of physical nodes
     * @param pNodeList, all physical hosts in the cluster
     */
    public void construct(List<PNode> pNodeList){
        this.numOfpNode = pNodeList.size();
        this.uplinkCapacity = new Long[numOfpNode];
        this.downlinkCapacity = new Long[numOfpNode];
        this.availableDownlinkBw = new Long[numOfpNode];
        this.availableUplinkBw = new Long[numOfpNode];
        this.pNodes = new PNode[numOfpNode];

        // init the bipartite shuffle graph
        for (int i = 0; i < pNodeList.size(); i++) {
            PNode pNode = pNodeList.get(i);
            uplinkCapacity[i] = pNode.getAggregateBw();
            downlinkCapacity[i] = pNode.getAggregateBw();
            availableUplinkBw[i] = pNode.getAggregateBw();
            availableDownlinkBw[i] = pNode.getAggregateBw();
            pNodes[i] = pNode;
        }

        // init the book-keeping structure
        for (int i = 0; i < pNodeList.size(); i++) {
            PNode pNode = pNodeList.get(i);
            List<VNode> vNodes = pNode.getAllVNodes();

            for (int j = 0; j < vNodes.size(); j++) {
                VNode vNode = vNodes.get(j);
                this.vnodeIdToIdx.put(vNode.getHostName(), i);
                this.vnodeIdToPNode.put(vNode.getHostName(), pNode);
            }
        }
    }

    public void addFlow(Flow flow){
        FlowRun newFlow = new FlowRun(flow.getFlowid(),
                this.vnodeIdToIdx.get(flow.getSrcHostname()),
                this.vnodeIdToIdx.get(flow.getDstHostname()),
                flow.getBwShare(), flow.getFlowSize(), this);
        this.flowActors.put(flow.getFlowid(), newFlow);
    }

    public void delFlow(long flowId){
        //LOG.info("Deleting flow " + flowId);
        FlowRun f = flowActors.remove(flowId);
        if(f == null){
            LOG.error("Can't delete non-exist flow " + flowId);
        } else {
            //LOG.info("Deallocating bw of " + flowId);
            deallocate(f.getSrcIdx(), f.getDstIdx(), f.getRtBw());
            f.stop();
        }
        LOG.debug("Flow " + flowId + " deleted");
        /*
        if(f == null){
            LOG.error("Can't delete non-exist flow " + flowId);
        } else {
            // deallocate bw
            // remove from flow book-keeping
            f.exit();
            flowActors.remove(flowId);
        }
        */
    }

    public void startFlow(long flowId) {
        //  Start to simulate a flow run
        FlowRun f = getRunningFlow(flowId);
        if(f == null){
            LOG.error("Can't start non-exist flow " + flowId);
        } else {
            // first dibs on the bw
            allocate(f.getSrcIdx(), f.getDstIdx(), f.getRtBw());
            // then run
            f.run();
        }
    }

    // TODO: code review
    //  - really need this?
    public void suspendFlow(long flowId){
        //  Suspend a flow
        FlowRun f = getRunningFlow(flowId);
        if(f == null){
            LOG.error("Can't suspend non-exist flow " + flowId);
        } else {
            deallocate(f.getSrcIdx(), f.getDstIdx(), f.getRtBw());
            f.suspend();
        }
    }

    // TODO: code review
    //  - really need this?
    public void resumeFlow(long flowId){
        //  Resume a flow
        FlowRun f = getRunningFlow(flowId);
        if(f == null){
            LOG.error("Can't resume non-exist flow " + flowId);
        } else {
            allocate(f.getSrcIdx(), f.getDstIdx(), f.getRtBw());
            f.resume();
        }
    }

    public void throttle(long flowId, long rateLimit){
        //  Adjust the bw share of a flow
        FlowRun f = getRunningFlow(flowId);
        if(f == null){
            LOG.error("Can't throttle non-exist flow " + flowId);
        } else {
            if(rateLimit < f.getRtBw()){
                // reduce the real time bw
                deallocate(f.getSrcIdx(), f.getDstIdx(), f.getRtBw()-rateLimit); // deallocate the delta part
            } else if(rateLimit > f.getRtBw()){
                // increase the real time bw
                allocate(f.getSrcIdx(), f.getDstIdx(), rateLimit-f.getRtBw()); // allocate the delta part
            }
            f.throttle(rateLimit); // notify actor updating his real-time rate
        }
    }

    public void shutdown(){
        // TODO
        // Gracefully shutdown
    }

    public Queue<FlowFinishEvent> getEventQueue() {
        Queue<FlowFinishEvent> ret = null;
        synchronized (this.flowFinishEventQueue){
            ret = this.flowFinishEventQueue;
            this.flowFinishEventQueue = new LinkedList<FlowFinishEvent>();
        }
        return ret;
    }

    public void addFlowFinishEvent(long flowId){
        synchronized (this.flowFinishEventQueue){
            this.flowFinishEventQueue.add(new FlowFinishEvent(flowId));
        }
    }
    // Called by

    // Query interfaces

    public int getNumOfpNode() {
        return numOfpNode;
    }

    public void setNumOfpNode(int numOfpNode) {
        this.numOfpNode = numOfpNode;
    }

    /**
     * Get available aggregate bandwidth on which the given vnode is
     */
    public long getAvailableUpBw(VNode src) throws Exception {
        return availableUplinkBw[vnodeIdToIdx.get(src.getHostName())];
    }

    /**
     * Get available aggregate bandwidth on which the given vnode is
     */
    public long getAvailableDownBw(VNode dst) throws Exception {
        return availableDownlinkBw[vnodeIdToIdx.get(dst.getHostName())];
    }

    /**
     * Get available aggregate bandwidth on which the given vnode is
     * TODO: need this?
     */
    public long getAvailableDownBw(PNode src) throws Exception{
        //TODO: FIXME
        return -1;
    }

    public void printState(PNode pNode){
        int target = -1;
        for (int i = 0; i < pNodes.length; i++) {
            if(pNodes[i].getDpip() == pNode.getDpip()){
                target = i;
                break;
            }
        }
        if(target == -1) return;
        LOG.info("Available: " + availableUplinkBw[target]);
        for(Map.Entry<Long, FlowRun> flow : flowActors.entrySet()){
            if(flow.getValue().isStarted())
                LOG.info("flow " + flow.getKey() + " takes " + flow.getValue().getRtBw());
        }
    }
    //
    // Inner classes
    //

    /*
    private class Flow extends pfsimulator.Flow{
        public Flow(pfsimulator.Flow thinFlow){
            super(thinFlow.getFlowid(), thinFlow.getSrc(),
                    thinFlow.getDst(), thinFlow.getBwShare());
        }
    }
    */

}
